/* Copyright 2011 Taobao.com

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */


#include "repsync.h"
#include <mysql/errmsg.h>
#include <string.h>

#undef sem_wait
#undef sem_post
#undef assert
#include <assert.h>

pthread_mutex_t     *slave_conn_mutex;
pthread_mutex_t     repsync_logfile_mutex;
static setting_st   setting;
master_info_st      *glob_master    =   NULL;
slave_info_st       *glob_slave     =   NULL;
pk_map_t            *pk_map;

static void get_log_dir(const char *, char *path, size_t len);
static int load_posfile(master_info_st *master);
void* do_master_binlog(void *);
static size_t write_binlog(master_info_st *master, const char *,size_t len, my_off_t pos);
static int open_binlog(const char *binlog_path);
static int update_posfile_binlog(master_info_st *master);
static int close_binlog(int fd);
static int daemon();
void *do_slave_relaylog(void *arg);

static void get_curr_time(char *result) {
	time_t now=time(0);
	tm *tnow = localtime(&now);
	sprintf(result,"%d-%d %d:%d:%d", 
			tnow->tm_mon+1,
			tnow->tm_mday,
			tnow->tm_hour,
			tnow->tm_min,
			tnow->tm_sec);
}

static char time_buf[256];                             
#define my_printf(fmt, ...)                         \
	pthread_mutex_lock(&repsync_logfile_mutex);     \
	fflush(fp_stdout);                              \
	bzero(time_buf,256);                            \
	get_curr_time(time_buf);                        \
	printf("%s %s:%d %d : ",time_buf , __FILE__,    \
            __LINE__ , (uint)pthread_self());       \
	printf(fmt, ##__VA_ARGS__);                     \
	printf("\n");                                   \
	fflush(fp_stdout);                              \
	pthread_mutex_unlock(&repsync_logfile_mutex)

#define PRINTF_BINLOG_PATH(path, master)    \
        snprintf(path, PATH_MAX, "%s/%s", setting.base_dir, master->mysql_info->name);                               


#define PRINTF_DUMPSQL_PATH(path, master)    \
        snprintf(path, PATH_MAX, "%s/%s/%s.dump", setting.base_dir,  master->mysql_info->name, master->mysql_info->table);                               


#define EXIT_THREAD(mysql)              \
   my_printf("thread exist ...");       \
   mysql_close(mysql);                  \
   setting.running = 0;                 \
   pthread_exit(NULL);


int connect_mysql(MYSQL *mysql, mysql_info_st *mysql_info)
{
	int count=0;
	while (1) {
        if (!(mysql_real_connect(mysql, 
                        mysql_info->host,
                        mysql_info->user,
                        mysql_info->password,
                        mysql_info->database, 
                        mysql_info->port, 
                        NULL, 
                        CLIENT_MULTI_STATEMENTS|CLIENT_MULTI_RESULTS|MYSQL_OPT_RECONNECT))) {
            DEBUG("Couldn't connect to server %s:%d:%s:%s,now sleep for 60s,then reconnect\n", 
                    mysql_info->host,
					mysql_info->port,
                    mysql_info->user,
                    mysql_info->password);
			if (count++ == 10)
				return -1;
			sleep(60);
		}
		else{
            mysql_options(mysql, MYSQL_SET_CHARSET_NAME, "gbk");
            mysql_query(mysql, "set names 'gbk'");
			mysql->reconnect = 1;
			return 0;
		}
	}
}


int repsync_kill()
{
	FILE *fp;
	const char *path = setting.pidfile;
	if (!path && (fp=fopen(path,"r")) == NULL) {
        DEBUG("open pid file error: %s", path);
		return -1;
	}
	char buf[60];
	if (!fp && fgets(buf,64,fp)==NULL) {
		DEBUG("error while fgets");
		return -1;
	}
	uint pid=atoi(buf);
	printf("pid=%d KILLED\n",pid);
	if (unlink(path) != 0) {
		DEBUG("unlink pidfile error");
	}
	kill(pid,SIGKILL);
	return 0;
}

static void repsync_clean(int sig)
{
	if (unlink(setting.pidfile) != 0) {
		DEBUG("unlink pidfile error");
	}
	int i;
	for (i=0; i <setting.n_master; i++) {
		close_binlog(glob_master[i].fd);
		if (close( glob_master[i].pos_fd )) {
			DEBUG("close the pos file error:%s", strerror(errno));
		} else {
			glob_master[i].pos_fd=-1;
        }

        free(*(void**)&glob_master[i].binlog_start);
        free(*(void**)&glob_master[i].binlog_cur);

	}
	free(glob_master);

    for (i=0; i<setting.n_slave; i++) {
        free(glob_slave[i].mysql_info);
        free(glob_slave[i].mysql);
    }

    free(glob_slave);
    free(slave_conn_mutex);
    //free(setting.pidfile);
    //free(setting.logfile);
    //free(setting.base_dir);
    //free(setting.cnf_file);
	exit(0);
}

/* target could be split by ',' */
int my_strstr(const char *query, char *target) {
    
    char *p;
    char *split=",";
     
    p = strtok(target, split);
    while (p) {
        if (strstr(query, p)) {
            return 0;
        }

        p=strtok(NULL, split);
    }

    free(target);
    return -1;
}

void dump_diff_data(FILE *fd, const char *query) {
    // currently disable it in the case of exhausting disk space
    #if 0
    if (EOF == fputs(query, fd) ||
        (fflush(fd) != 0)) {
        DEBUG("dump diff data failed.");

        return;
    }

    fputs("\n", fd);
    #endif
}


int find_first_key_pos(const char *query, const char *key) {
    char *p;
    
    int key_len = strlen(key);
    char *low_key = (char*)malloc(key_len+1);
    low_key[key_len] = '\0';
    char *upp_key = (char*)malloc(key_len+1);
    upp_key[key_len] = '\0';
    for (int i=0; i<key_len; i++) {
        low_key[i] = tolower(key[i]);
        upp_key[i] = toupper(key [i]);
    }

    int len = 0;

retry:
    p = strstr(query+len, upp_key);
    if (!p)
        p = strstr(query+len, low_key);

    if (!p) {
        if (strcasecmp(query, "begin") == 0) 
            return -1;

        DEBUG("can not find key=%s in query=%s", key, query);
        return -1;
    }


    len = p - query;

    while (!isspace(*(p-1)) || !isspace(*(p+strlen(key)))) {
        len++;
        goto retry;
    }

    free(low_key);
    free(upp_key);

    return len;
}
                                
/* Insert situations:
 * 1. insert into tbl values('a', 'b');
 * 2. insert into tbl(nick, comment) values('a', 'b');
 * 3. insert into tbl (id, nick, comment) values(null, 'a', 'b');
 */
char *convert_autoinc_id_with_val(const char *query, char *id_map, unsigned long long intval) { 

    int pos = find_first_key_pos(query, "into");
    if (pos == -1) {
        free(id_map);
        return NULL;
    }
    
    char *new_query = NULL; 
    const char *p = query+pos+strlen("into");

    while (isspace(*p )) { /* until hit table name */
        p++;
    }
    int offset = 0;
    while (!isspace(*(p+offset)) && (*(p+offset) != '(')) {
        offset++;
    }
    if (isspace(*(p+offset))) {
        offset++;
        while (isspace(*(p+offset))) {
            offset++;
        }
    }

    
    if (*(p+offset) == '(') {
        new_query = (char*)malloc(ITEM_LEN);
        memset(new_query, 0, ITEM_LEN);

        const char *p1 = strstr(query, "(");
        const char *p2 = strstr(p1+1, "(");
       
        offset = 1;
        while (isspace(*(p1+offset))) {
            offset++;
        }
        // find the table mapped id
        // table1:pk1,table2:pk2,...
        char *ptok  = NULL;
        char *potok, *potok2;
        char *ps = NULL;
        char *search_table, *low_pk, *upp_pk;
        for (int i=0; i<MAX_PK; i++) {
           if (pk_map[i].valid == 0)
                return NULL; /*don't need convert*/

            ps=strstr(query, pk_map[i].upp_table);
            if (ps == NULL)
                ps=strstr(query, pk_map[i].low_table);
            if (ps != NULL && ps < p1) {
                low_pk = pk_map[i].low_pk;
                upp_pk = pk_map[i].upp_pk;
                break;
            }
        }


       
        if ((strncmp(p1+offset, low_pk, strlen(low_pk)) == 0) || 
            (strncmp(p1+offset, upp_pk, strlen(upp_pk)) == 0))
        {
            const char *p3 = strstr(p2, ",");

            snprintf(new_query, p2-query+2, "%s", query);
            snprintf(new_query+strlen(new_query), 12, "%d,", intval);
            snprintf(new_query+strlen(new_query), ITEM_LEN, "%s", p3+1);

        } else {

            snprintf(new_query, p1-query+2, "%s", query);
            snprintf(new_query+strlen(new_query), strlen(low_pk)+2, "%s,", low_pk);
            snprintf(new_query+strlen(new_query), p2-p1+1, "%s", p1+1);
            snprintf(new_query+strlen(new_query), 12, "%d,", intval);
            snprintf(new_query+strlen(new_query), ITEM_LEN, "%s", p2+1);
        }


    } else if (*(p+offset) == 'v' || *(p+offset)=='V') {
        //FIXME: don't support such case yet:
        // insert into tab2 values(NULL);
        // where only pk for tab2
        //
        // For insert into tab values(0, 'xyz') OR
        //     insert into tab values(pk_id, 'xyz')
        //
        // DO NOTHING, just let new_query return NULL
#if 0
        const char *p1 = strstr(query, "(");
        const char *p2 = strstr(p1+1, "(");

        snprintf(new_query, p2-query+1, "%s,", query);
        snprintf(new_query+strlen(new_query), 12, "%d,", intval);
        snprintf(new_query+strlen(new_query), ITEM_LEN, "%s", p2+1);
#endif
        // same as return NULL;
    }
    free(id_map);
   
    return new_query;
}


/* replace the database/table for transfer rules in rep.cnf
 * currently only support db convert while not table
 * */
char* transform_query(const char *query, const char *src_str, 
        const char *des_str, const char *tab_str, int *to_free) {

    char *buffer = NULL;
    char *p;
    *to_free = 0;
    
    if (strcmp(src_str, des_str) == 0)
        return buffer;

    char db_str[150]; 
    snprintf(db_str, 50, "%s.", src_str);
    if ((p = strstr(query, db_str))) {
        // replace database name
        // TODO: handle join situation:
        // db1.tab.pk = db2.tab.fk and so on.
        // looping replace 


        //TODO: fix this situation. FIXED
        //"insert into repsync.t1 values (111)", src_str=0x4f7fc14 "sbtest", des_str=0x4f47d30 "repsync",
        //
        buffer = (char*)calloc(ITEM_LEN, sizeof(char));
        assert(buffer != NULL);
        *to_free = 1;
        strncpy(buffer, query, p-query);
        buffer[p-query] = '\0';
        sprintf(buffer+(p-query), "%s%s", des_str, p+strlen(src_str));

    } else {

        return buffer;

        //remove the codes below later as we selected the db during connection
        //
        // replace table name to database.table
        // TODO: handle join situation, multi-table are used

        if (tab_str && (p = strstr(query, tab_str))) {
            buffer = (char*)calloc(ITEM_LEN, sizeof(char));
            assert(buffer != NULL);
            *to_free = 1;

            strncpy(buffer, query, p-query);
            buffer[p-query] = '\0';
            sprintf(buffer+(p-query), "%s.%s;", des_str, p);
        }
    }

    return buffer;
}



int process_mysql_query(MYSQL *mysql, const char *query, master_info_st *master) {

    int ret = 0;
    
    if (setting.ignore && (my_strstr(query, strdup(setting.ignore)) == 0))
         return 0;

#ifdef USE_LOCK_FOR_QUERY
    if (setting.n_master > setting.n_slave)
        pthread_mutex_lock(&slave_conn_mutex[slave_index]);
#endif

    // keep the correct query into separated files for providing services

    if (setting.dump_diff == 1) {
        char comm[ITEM_LEN];
        memset(comm, 0, ITEM_LEN);

        typedef enum e_opr{INSERT, UPDATE, DELETE, UNKNOWN} e_opr;
        e_opr opr=UNKNOWN, status;
        char *sret;

        if (strstr(query, "update ") || strstr(query, "delete ")  || 
            strstr(query, "UPDATE ") || strstr(query, "DELETE ") ) {

            char tabname[50];
            char *tp, *p1, *p2;
            if (strstr(query, "update ") || strstr(query, "UPDATE ")) {
                if (strstr(query, "update ")) {
                    tp = strstr(query, "update");
                    p1 = strstr(tp+strlen("update"), " ");
                 } else {
                    tp = strstr(query, "UPDATE");
                    p1 = strstr(tp+strlen("UPDATE"), " ");
                 }
   
                int offset=1;
                while (*(p1+offset) == ' ')
                    offset++;
                p2 = strstr(p1+offset, " ");                
                snprintf(tabname, p2-p1+1, "%s", p1);

            } if (strstr(query, "delete ") || strstr(query, "DELETE ")) {
                if (strstr(query, "from")) {
                    tp = strstr(query, "from");
                    p1 = strstr(tp+strlen("from"), " ");
                } else {
                    tp = strstr(query, "FROM");
                    p1 = strstr(tp+strlen("FROM"), " ");
                }

                int offset=1;
                while (*(p1+offset) == ' ')
                    offset++;
                p2 = strstr(p1+offset, " ");                
                snprintf(tabname, p2-p1+1, "%s", p1);
            } else {
                // don't need this way as mysql_info->table might contain table pattern
                // p1 = *(char**)&master->mysql_info->table; 
                // snprintf(tabname, 50, " %s", p1);
            }
 

            snprintf(comm, ITEM_LEN, "select * from %s ", tabname) ;

            char *p = strstr(query, "where "); // TODO: tab format for left or right
            if (p == NULL) 
                p = strstr(query, "WHERE ");
            if (p == NULL) {
                DEBUG("failed to locate the where condition for update/delete: %s", query);
                return 0;
            }
            snprintf(comm+strlen(comm), ITEM_LEN-strlen(comm), "%s ", p);

            MYSQL *dump_mysql = NULL;
            if (strstr(query, "delete ") || (strstr(query, "DELETE "))) {
                dump_mysql = mysql;
                opr=DELETE;
            }
            else if (strstr(query, "update ") || strstr(query, "UPDATE "))  {
                dump_mysql = &master->mysql;
                opr=UPDATE;
            } else if (strstr(query, "insert ") || strstr(query, "INSERT "))  {
                dump_mysql = &master->mysql;
                opr=INSERT;
            }

            if (setting.forward_slave == 1) {
                // temperary solution for speedup slave's replication
                // suggested by junda
                dump_mysql = mysql;
            }
            if (setting.debug_sql) {
                DEBUG("dump sql=%s", comm); 
            }

            if ((ret=mysql_query(dump_mysql, comm)) != 0) {
                DEBUG("%s %s %s:\n query error: %s \n %s",  
                        master->mysql_info->host,  
                        master->mysql_info->database, 
                        master->mysql_info->table, 
                        mysql_error(dump_mysql), 
                        query);
                DEBUG("dump sql=%s", comm); 

            } else {
                MYSQL_RES* res = NULL;
                if (!(res = mysql_store_result(dump_mysql))) {
                    DEBUG("Couldn't get result from %s", mysql_error(dump_mysql));
                    EXIT_THREAD(dump_mysql);
                } else if (res->row_count > 0) {
                    MYSQL_ROW row=mysql_fetch_row(res);
                    char row_buffer[ITEM_LEN]; // enlarge?
                    memset(row_buffer, 0, ITEM_LEN);

#define GET_ENUM_STRING(e) case e: sret = #e; break;
                    switch (opr) {
                        GET_ENUM_STRING(INSERT);
                        GET_ENUM_STRING(UPDATE);
                        GET_ENUM_STRING(DELETE);
                        default:
                        break;
                    }

                    snprintf(row_buffer+strlen(row_buffer), ITEM_LEN, "%s:", sret);
                    for (int i=0; i<res->field_count; i++) {
                        snprintf(row_buffer+strlen(row_buffer), ITEM_LEN, "%s ", row[i]); 
                    }

                    dump_diff_data(glob_master[master->group_head_index].dump_fd, row_buffer);
                }
                mysql_free_result(res);
            }
        } else if (strstr(query, "insert ") || strstr(query, "INSERT ") ) {
            // handle insert: extract "insert into table values (/pattern/);"

            opr = INSERT;
            char *tp = strstr(query, "values");
            if (tp == NULL)
                tp = strstr(query, "VALUES");

            char *p = strstr(tp, "("); // first '('
            char *p0 = p+1;
            char *p1 = strstr(p, ")");

            while (1) {
                p = p1+1;
                if (!p || !strstr(p, ")")) 
                    break;
                else
                    p1 = p;
            }

#define GET_ENUM_STRING(e) case e: sret = #e; break;
            switch (opr) {
                GET_ENUM_STRING(INSERT);
                GET_ENUM_STRING(UPDATE);
                GET_ENUM_STRING(DELETE);
                default:
                break;
            }

            snprintf(comm, ITEM_LEN, "%s:", sret);
            int len = strlen(comm);
            snprintf(comm+len, ITEM_LEN, "%s", p0);
            comm[p1-p0+len]='\0';
            dump_diff_data(glob_master[master->group_head_index].dump_fd, comm);
 
        }
    }

    
    if (setting.apply_binlog) {
        if (setting.debug_sql) {
            DEBUG("sql=%s", query); 
        }

retry:
        if ((ret=mysql_query(mysql, query)) != 0) {
            DEBUG("%s %s %s:\n query error: %s \n %s",  
                    master->mysql_info->host,  
                    master->mysql_info->database, 
                    master->mysql_info->table, 
                    mysql_error(mysql), 
                    query);

            // error handler: broken connection or real errror
            //
            if (mysql_errno(mysql) == CR_SERVER_LOST ||
                mysql_errno(mysql) == CR_SERVER_GONE_ERROR
                ) {

                DEBUG("slave connection abort as %s, reconnection...", mysql_error(mysql));

                mysql_close(mysql);
                mysql_init(mysql);
                if (connect_mysql(mysql, glob_slave[master->slave_index].mysql_info) != 0)
                    goto retry;

                // reset server_id if slave host is changed
                //mysql->mysql_info->server_id = get_server_id(mysql);
                
            } else if (mysql_errno(mysql) != 1062) {

                // Error: 1062 SQLSTATE: 23000 (ER_DUP_ENTRY)
                // Message: Duplicate entry '%s' for key %d

                DEBUG("error: insert failed while not duplicate key issue");
                if (setting.pattern == 1) {
                    EXIT_THREAD(mysql);

                }

            }


            // restore the auto increase to max id in INSERT statement
          

            if (setting.autoinc_id &&
                (strstr(query, "insert ") || strstr(query, "INSERT ") )) {
             
#if 0
                // insert into table values 
                // insert into  table (
                // pattern: into xxx, end with '('|' '
                
                char *p = strstr(query, "into ");
                if (!p)
                    p = strstr(query, "INTO ");
                if (!p) {
                    DEBUG("abnormal query of %s, without into|INTO key", query);
                    return -1;
                }

                p += 5;
                int offset = 0;

                while (*p == ' ') {
                    p++;
                }
                while ((*(p+offset) != ' ')  && (*(p+offset) != '(')) {
                    offset++;
                }
                
                char table_name[150];
                memset(table_name, 0, 150);
                snprintf(table_name, offset+1, "%s", p);

                char set_query[ITEM_LEN];
                memset(set_query, 0, ITEM_LEN);
                snprintf(set_query, ITEM_LEN, "select max(%s) from %s",
                        setting.autoinc_id, table_name);



//                unsigned long long last_id = 0;
//                last_id = mysql_insert_id(mysql);
//                int xid = mysql_insert_id(mysql);
//                if ((ret=mysql_query(mysql, "select LAST_INSERT_ID()")) != 0) {
                  if ((ret=mysql_query(mysql, set_query))) { 
//                    DEBUG("query error for SELECT LAST_INSERT_ID() table=%s error=%s ",
                    DEBUG("query error for %s table=%s error=%s", set_query, 
                            table_name,
                            mysql_error(mysql));
                } else {

                    MYSQL_RES *res;

                    if (!(res = mysql_store_result(mysql))) {
                        DEBUG(" failed to restore result ... ");
                    } else {
                        MYSQL_ROW row;
                        if ((row=mysql_fetch_row(res))) {
                            uint32_t id = atoi(row[0]);
                            mysql_free_result(res);
                            snprintf(set_query, ITEM_LEN, "ALTER TABLE %s AUTO_INCREMENT = %d", table_name, id);
                            if ((ret=mysql_query(mysql,  set_query)) != 0) {
                                DEBUG("set auto_increment failed: %s", mysql_error(mysql));
                            } else {
                                DEBUG("set auto increment for %s with %d", table_name, id);
                            }
                        }
                    }
                }
                
                #endif

            } 

        }
    }

#ifdef USE_LOCK_FOR_QUERY
    if (setting.n_master > setting.n_slave)
        pthread_mutex_unlock(&slave_conn_mutex[slave_index]);
#endif

    return ret;
}


static int load_cnf_info() {
	int i=0;
	FILE *fp;
    const char* cnf_file = setting.cnf_file;
	char buf[MAX_STRING_LEN];

	if ( (fp=fopen(cnf_file,"r")) == NULL ) {
		DEBUG("rep.cnf file does not exist");
		if ( (fp=fopen(cnf_file,"w")) == NULL ) {
			DEBUG("create rep.cnf error: %s", strerror(errno));
			return -1;
		}

		char info[]="#[master_slave_mapping_info]    \n\n\
#master|ip|port|name|db|table|user|pwd|binlog|path\n\
#slave|ip|port|name|db|table|user|pwd\n\
#Example\n\
master|10.232.31.70|3306|M1|0|0|repsync|repsync|0|0\n\
slave|10.232.31.70|3307|S1|0|0|repsync|repsync||\n";

		if (fputs(info,fp)==EOF) {
			DEBUG("write into rep.cnf error");
			return -1;
		}

		DEBUG("cnf template file %s is created, please edit it and rerun", cnf_file);
		return -1;
	}
	if ( (fseek(fp,0,SEEK_SET)) != 0 ) {
		DEBUG("lseek rep.cnf file error");
		return -1;
	}

	while (fgets(buf,MAX_STRING_LEN,fp) != NULL) {
		if (buf[0] != '#' && strcmp(buf,"\n")) {
            if (strncmp(buf, "master", strlen("master")) == 0)
                setting.n_master++;
             else if (strncmp(buf, "slave", strlen("slave")) == 0)
                setting.n_slave++;
        }
	}

    if (setting.n_master < 1 || setting.n_slave < 1) {
        DEBUG("no valid information in cnf_info, \
                please check it: n_master=%d, n_slave=%d\n", 
                setting.n_master, setting.n_slave);
        return -1;
    }

	if ( (fseek(fp,0,SEEK_SET)) != 0 ) {
		DEBUG("lseek cnf_info file error");
		return -1;
	}

    glob_master= (master_info_st *)calloc(setting.n_master, 
            sizeof(master_info_st)); 
    assert(glob_master != NULL);
    glob_slave= (slave_info_st *)calloc(setting.n_slave, 
            sizeof(slave_info_st)); 
    assert(glob_slave != NULL);

    
    int master_index    = 0;
    int slave_index     = 0;
    int group_masters   = 0;
    int group_index     = 0;
    int group_head_index     = 0;

	while (fgets(buf,MAX_STRING_LEN,fp) != NULL) {
		if (buf[0] == '#' || !strcmp(buf,"\n"))
			continue;

        mysql_info_st *mysql_info = (mysql_info_st *)malloc(
                sizeof(mysql_info_st)); 
        assert(mysql_info != NULL);
        mysql_info->server_id = -1;

		char *p, *p_buf;
		buf[strlen(buf)-1] = '\0';
		p_buf = buf;

        int is_slave = 0;
        const char *binlog_start = NULL;
        char host[MAX_STRING_LEN];
        int j = 0;
        int binlog_pos = 0;

#define CONVERT_STRDUP(p) \
    (strcmp(p, "0") == 0) ? NULL : strdup(p)

		for (;;j++) {
			p=strtok(p_buf,"|");
			if (p==NULL)
				break;
			switch (j)
			{
                case 0:
                    if (strcmp((p), "slave") == 0) {
                        is_slave = 1;
                    }
                    break;
				case 1: 
                    mysql_info->host = strdup(p); 
                    break;
				case 2:
                    mysql_info->port = atoi(p);
                    break;
				case 3:
                    mysql_info->name = strdup(p);
                    break;
				case 4: 
                    mysql_info->database = CONVERT_STRDUP(p);
                    break;
                case 5: 
                    mysql_info->table = CONVERT_STRDUP(p);
                    break;

                case 6:    
                    mysql_info->user = strdup(p);
                    break;
				case 7: 
                    mysql_info->password= strdup(p);
                    break;
				case 8: 
                    binlog_start = CONVERT_STRDUP(p);
                    break;
                case 9: 
                    binlog_pos = atol(p);
                    if (binlog_pos == 0)
                        binlog_pos = LOG_POS_INIT;
                    break;
 
                case 10:
                    if (setting.base_dir == NULL) {
                        setting.base_dir = CONVERT_STRDUP(p);
                        if (setting.base_dir == NULL) {
                            setting.base_dir = "./repsync_data";
                        }
                    }
                    break;
            }
			p_buf = NULL;
		}

        if (is_slave) {
            is_slave = 0;
            glob_slave[slave_index].mysql_info = mysql_info;
            glob_slave[slave_index].n_master   = group_masters;
            group_masters = 0;
            slave_index++;
            group_head_index = master_index;
        } else {
            glob_master[master_index].mysql_info        = mysql_info;
            glob_master[master_index].binlog_start      = binlog_start;
            glob_master[master_index].binlog_pos        = binlog_pos;
            glob_master[master_index].slave_index       = slave_index;
            glob_master[master_index].index             = master_index;
            glob_master[master_index].group_index       = group_masters;
            glob_master[master_index].group_head_index  = group_head_index;
            glob_master[master_index].tid               = 0;
            glob_master[master_index].fd                = -1;
            glob_master[master_index].description_event = NULL;
            glob_master[master_index].stats_binlog_write= 0;
            glob_master[master_index].pos_fd            = -1;

            master_index++;
            group_masters++;
        }

        i++;
    }

    assert(setting.n_master >= setting.n_slave); 
    assert(setting.n_master == master_index+1);
    assert(setting.n_slave == slave_index+1);
 
    const char *base_host = glob_master[0].mysql_info->host;
    const char *base_db = glob_master[0].mysql_info->database;
    const char *base_name = glob_master[0].mysql_info->name;
    int base_port = glob_master[0].mysql_info->port;   
    int base_index  = 0;
    int n_slides    = 1;
    int same_group  = 1;
    slave_index = glob_master[0].slave_index;

    for (i=0; i<setting.n_master; i++) {
        // judge <host, port, db> for an instance
        // could fuzzy to <host, port> if needed

        if (strcmp(glob_master[i].mysql_info->host, base_host) !=0 ||
            (!glob_master[i].mysql_info->database ||
             glob_master[i].mysql_info->database && 
             strcmp(glob_master[i].mysql_info->database, base_db) !=0) ||
            (glob_master[i].mysql_info->port != base_port) 
            ) {
            
            same_group = 0;
        }

        if (same_group == 1 && setting.force_group == 1) {
            if (strcmp(glob_master[i].mysql_info->name, base_name) != 0)
                same_group = 0;
        }


        /*if (strcmp(glob_master[i].mysql_info->host, base_host) !=0 ||
            strcmp(glob_master[i].mysql_info->database, base_db) !=0 ||
            (glob_master[i].mysql_info->port != base_port) 
            ) { */
        if (same_group == 0) {    

            base_host = glob_master[i].mysql_info->host;
            base_db = glob_master[i].mysql_info->database;
            base_name = glob_master[i].mysql_info->name;
            base_port = glob_master[i].mysql_info->port;
            n_slides++;

            if (slave_index != glob_master[i].slave_index) {
                n_slides--;
                glob_slave[slave_index].tids = (pthread_t*) malloc(
                        sizeof(pthread_t) * n_slides);
                glob_slave[slave_index].n_slides = n_slides;
                slave_index = glob_master[i].slave_index;
                n_slides = 1;
            }
            
            base_index = i;
                same_group = 1;
        }
        
        glob_master[base_index].n_instance++;
    }

    glob_slave[slave_index].tids = (pthread_t*) malloc(
            sizeof(pthread_t) * n_slides);
    glob_slave[slave_index].n_slides = n_slides;


    DEBUG("save_binlog=%d, apply_binlog=%d, daemon=%d, debug_sql=%d\n", 
            setting.save_binlog, setting.apply_binlog, setting.daemon, 
            setting.debug_sql);

    return 0;
}


uint32_t get_server_id(MYSQL *mysql) {
	MYSQL_RES *res;

	if (mysql_query(mysql, "select @@server_id;")) {
		DEBUG("Query faild (%s)", mysql_error(mysql));
        EXIT_THREAD(mysql);
	}

	if (!(res = mysql_store_result(mysql))) {
		DEBUG("Couldn't get result from %s", mysql_error(mysql));
        EXIT_THREAD(mysql);
	}

	MYSQL_ROW row=mysql_fetch_row(res);
	uint32_t id = atoi(row[0]);
    mysql_free_result(res);

    return id;
}


static int send_binlog_dump(MYSQL *mysql, master_info_st *master, uint32 server_id) {

	uchar buf[MAX_STRING_LEN];
	ulong len;
	uint logname_len;

	const char *logname = master->binlog_cur;
	NET* net = &mysql->net;

	uint32 n=4; //original set start to n(4)
	int4store(buf, (unsigned long) master->binlog_pos);
	int2store(buf + 4, 0);
    int4store(buf + 6, server_id);
    
    //FIXME: unset to LOG_POS_INIT while read from files
    if (master->binlog_pos != LOG_POS_INIT)
        master->binlog_pos = LOG_POS_INIT;

	size_t tlen = strlen(logname);
	if (tlen > UINT_MAX) {
		DEBUG("Log name too long.");
		return -1;
	}
	logname_len = (uint) tlen;
	memcpy(buf + 10, logname, logname_len);

	if (simple_command(mysql, COM_BINLOG_DUMP, buf, logname_len + 10, 1)) {
		DEBUG("Got fatal error sending the log dump command.");
		return -1;
	}
	return 0;
}

/**
  When reading a remote binlog, this function is used to grab the
  Format_description_log_event in the beginning of the stream.

  This is not as smart as check_header() (used for local log); it will
  not work for a binlog which mixes format. TODO: fix this.

  @retval DEBUG_STOP An error occurred - the program should terminate.
  @retval OK_CONTINUE No error, the program should continue.
 */
static Exit_status check_master_version(master_info_st *master, MYSQL *mysql)
{
	MYSQL_RES* res = 0;
	MYSQL_ROW row;
	const char* version;

	if (mysql_query(mysql, "SELECT VERSION()") ||
			!(res = mysql_store_result(mysql)))
	{
		DEBUG("Could not find server version: "
				"Query failed when checking master version: %s", mysql_error(mysql));
		return DEBUG_STOP;
	}
	if (!(row = mysql_fetch_row(res)))
	{
		DEBUG("Could not find server version: "
				"Master returned no rows for SELECT VERSION().");
		goto err;
	}

	if (!(version = row[0]))
	{
		DEBUG("Could not find server version: "
				"Master reported NULL for the version.");
		goto err;
	}

	delete master->description_event;
	switch (*version) {
		case '3':
			master->description_event= new Format_description_log_event(3);
			break;
		case '4':
			master->description_event= new Format_description_log_event(3);
			break;
		case '5':
			/*
			   The server is soon going to send us its Format_description 
               log event, unless it is a 5.0 server with 3.23 or 4.0 binlogs.
			   So we first assume that this is 4.0 (which is enough to read 
               the Format_desc event if one comes).
			 */
			master->description_event= new Format_description_log_event(4);
			break;
		default:
			master->description_event= NULL;
			DEBUG("Could not find server version: "
					"Master reported unrecognized MySQL version '%s'.", 
                    version);
			goto err;
	}
	if (!master->description_event || !master->description_event->is_valid())
	{
		DEBUG("Failed creating Format_description_log_event; out of memory?");
		goto err;
	}

	mysql_free_result(res);
	return OK_CONTINUE;

err:
	mysql_free_result(res);
	return DEBUG_STOP;
}


int create_master_thread(master_info_st * master)
{
	pthread_attr_t attr;
	int err;
	sigset_t mask, omask;

	sigemptyset(&mask);
	sigaddset(&mask, SIGINT);
	sigaddset(&mask, SIGQUIT);
	sigaddset(&mask, SIGHUP);
	assert(pthread_sigmask(SIG_SETMASK, &mask, &omask) == 0);

	if (pthread_attr_init(&attr) != 0) {
		DEBUG("pthread_attr_init");
		return -1;
	}

	if (pthread_attr_setstacksize(&attr, THREAD_STACK_SIZE) != 0) {
		DEBUG("pthread_attr_setsatcksize error");
		return -1;
	}
    
    DEBUG("creating master thread slot=%d/%d of %d tid=%d", master->index, master->group_index, master->slave_index, master->tid_index);
    pthread_t *tid = &glob_slave[master->slave_index].tids[master->tid_index];
	if ((err = pthread_create(tid, &attr, do_master_binlog, master)) 
        != 0) {
		DEBUG("unable to create an master thread :%s", strerror(err));
		return -1;
	}

	return 0;
}


int create_slave_thread(master_info_st *master)
{
	pthread_attr_t attr;
	int err;
	sigset_t mask, omask;

	sigemptyset(&mask);
	sigaddset(&mask, SIGINT);
	sigaddset(&mask, SIGQUIT);
	sigaddset(&mask, SIGHUP);
	assert(pthread_sigmask(SIG_SETMASK, &mask, &omask) == 0);

	if (pthread_attr_init(&attr) != 0) {
		DEBUG("pthread_attr_init");
		return -1;
	}

	if (pthread_attr_setstacksize(&attr, THREAD_STACK_SIZE) != 0) {
		DEBUG("pthread_attr_setsatcksize error");
		return -1;
	}
    DEBUG("creating slave thread slot=%d", master->index);

	if ((err = pthread_create(&master->tid, &attr, do_slave_relaylog, master)) 
        != 0) {
		DEBUG("unable to create an slave thread :%s", strerror(err));
		return -1;
	}

	return 0;
}


Exit_status process_event(master_info_st * master, Log_event *ev, size_t ev_len)
{
	Exit_status retval= OK_CONTINUE;
	const char * buf = ev->temp_buf;
	ev->temp_buf = NULL; 

	if (ev_len == 0) {
		delete ev;
        return OK_CONTINUE;
	}


	if (setting.save_binlog && 
        write_binlog(master, buf, ev_len, ev->log_pos) != ev_len)
		retval = DEBUG_STOP;
	else
		delete ev;

    return retval;
}


int reconnect_master(master_info_st *master, MYSQL *mysql)
{
	int ret = connect_mysql(mysql, master->mysql_info);

	if ( ret == 0) {
		char path[PATH_MAX];
        PRINTF_BINLOG_PATH(path, master);
		strcat(path,"/");
		strcat(path,master->binlog_cur);

		if (access(path, R_OK)==0) {
			if (unlink(path) != 0 && strlen(master->binlog_cur)) {
				DEBUG("unlink error:%s", strerror(errno));
				return -1;
			}
		}

		DEBUG("%s : start REGET Binlog : %s",
                master->mysql_info->name,
                master->binlog_cur);
        // Don't close binlog now as we assume it would continue to previous file pointer position
        /*
           if (master->fd != -1)
           close_binlog(master->fd);
           master->fd = open_binlog(path);
        */ 

		//update_posfile_binlog(master);
		return 0;
	}
	else
		return ret;
}


static Format_description_log_event* glob_description_event= NULL;


uint32_t generate_server_id(uint32_t server_id) {
    static uint32_t new_id = 0;

    return  __sync_add_and_fetch(&new_id, (server_id+10000));
}

/* src_list could be split by ',' */
uint32_t my_strncmp(char *src_list, const char *dst, uint32_t len) {

    uint32_t ret = 0;
    
    char *p;
    char *split=",";
    
    p = strtok(src_list, split);
    while (p) {
        if (strncmp(p, dst, len) == 0) {
            free(src_list);

            return 0;
        }

        p=strtok(NULL, split);
    }

            
    free(src_list);
    return 1;
}


/* establish one thread for each master while connecting to slave
 * it's thread safe 
 */
void *do_slave_relaylog(void *arg) {

	master_info_st * master = (master_info_st *)arg;
    //pthread_mutex_lock(&mutexs[slot]);
    int slot = master->index;
    const char *data;
    MYSQL *mysql = &glob_slave[master->slave_index].mysql[master->group_index];
    uint32_t count = 0;
    struct timeval tv_start, tv_end;
    unsigned long long tv_diff = 0;

    while (setting.running) { 
        while (sem_wait(&sems_empty[slot]) == -1 &&
                    EINTR == errno);

        data = buffer[slot][node[slot].head++];
        if (setting.debug_sql) {
            DEBUG("consume slot %d: %s", slot, data);
        }

        gettimeofday(&tv_start, NULL);
        process_mysql_query(mysql, data, master);
        gettimeofday(&tv_end, NULL);
        if (tv_end.tv_usec < tv_start.tv_usec) {            
            tv_diff += (1000000 + tv_end.tv_usec - tv_start.tv_usec);
            tv_diff += (tv_end.tv_sec-1 - tv_start.tv_sec)*1000000; 

        } else {
            tv_diff += (tv_end.tv_usec - tv_start.tv_usec);
            tv_diff += (tv_end.tv_sec - tv_start.tv_sec)*1000000; 
        }

        memset(buffer[slot][node[slot].head-1], 0, ITEM_LEN);

        if (node[slot].head >= SLOT_ITEMS)
            node[slot].head %= SLOT_ITEMS;
         
        if (++count >= N_STATS_ITV && 
                count % N_STATS_ITV == 0) {
            DEBUG("STATS: slot %d items=%d loops=%d rps=%d latency=%lu", 
                    slot, 
                    abs(node[slot].head-node[slot].tail),
                    ++(node[slot].loops%=99999999),
                    1000000*N_STATS_ITV/tv_diff,
                    tv_diff/N_STATS_ITV
                 );
             count = 0;
             tv_diff = 0;
             
         }
         sem_post(&sems_full[slot]);
    }
}

/* one thread access a group of slots 
 * it's thread safe 
 */
void put_data_to_buffer(int slot, const char *data) {
    
    while (sem_wait(&sems_full[slot]) == -1 &&
                    EINTR == errno );

    snprintf(buffer[slot][node[slot].tail++], ITEM_LEN, "%s", data);
    if (setting.debug_sql) {
        DEBUG("product slot %d: %s", slot, data);
    }
    
    if (node[slot].tail >= SLOT_ITEMS)
        node[slot].tail %= SLOT_ITEMS;

    sem_post(&sems_empty[slot]);
}

int find_slot_by_table(master_info_st *master, const char *query) {
    int i, k;
    const char      *p;
    mysql_info_st   *info;

    for (i=0, k=master->index; i<master->n_instance; i++, k++) {
        
        info = glob_master[k].mysql_info;
        if (setting.pattern == 1) {
            if (strstr(query, info->table))
                return k;
        } else {
            char tbl_name[150];
            char dbtbl_name[150];

            // for insert, insert into table_name ... values( ...
            // for update, update table_name set ...
            // for delete, delete from table_name where ...
            // TODO: create database/table handler
            int d = 0;
            const char *p, *p1;
            
            
            if (!(p = strstr(query, " into ")))
                p = strstr(query, " INTO ");
            if (p) { 
                p = strstr(query, "(");
                char tmp_str[150];
                memset(tmp_str, 0, 150);
                snprintf(tbl_name, 150, " %s", info->table);
                snprintf(dbtbl_name, 150, ".%s", info->table);

                snprintf(tmp_str, p-query+1, "%s", query);
                if ((p=strstr(tmp_str, tbl_name))) {
                    if (isspace(*(p+strlen(tbl_name))) || strncmp(p+strlen(tbl_name), "(", 1)==0)
                        return k;
                    else return -1;
                }
                else if ((p=strstr(tmp_str, dbtbl_name))) {
                    if (isspace(*(p+strlen(tbl_name))) || strncmp(p+strlen(tbl_name), "(", 1)==0)
                        return k;
                    else return -1;
                } else
                    return -1;
            }


            if (!(p = strstr(query, " set ")))
                p = strstr(query, " SET ");
                
            if (p) {
                snprintf(tbl_name, 150, " %s ", info->table);
                snprintf(dbtbl_name, 150, ".%s ", info->table);

                char tmp_str[150];
                memset(tmp_str, 0, 150);
                snprintf(tmp_str, p-query+2, "%s", query);
                if (strstr(tmp_str, tbl_name) || strstr(tmp_str, dbtbl_name)) 
                    return k;
                else
                    return -1;
            }


            if (!(p = strstr(query, " where ")))
                p = strstr(query, " WHERE ");
            if (p) {
                snprintf(tbl_name, 150, " %s ", info->table);
                snprintf(dbtbl_name, 150, ".%s ", info->table);

                char tmp_str[150];
                memset(tmp_str, 0, 150);
                snprintf(tmp_str, p-query+2, "%s", query);
                if (strstr(tmp_str, tbl_name) || strstr(tmp_str, dbtbl_name)) 
                    return k;
                else
                    return -1;
            }
        }
    }

    if (setting.debug_sql) {
        DEBUG("can not find table %s in query %s", info->table, query);
    }

    return -1;
}


void *do_master_binlog(void *arg)
{
	master_info_st * master = (master_info_st *)arg;
    master->trace_log_pos_offset = 54321;

    MYSQL *mysql = &glob_slave[master->slave_index].mysql[master->group_index];
    mysql_info_st *mysql_info = glob_slave[master->slave_index].mysql_info;
	const char *sid=master->mysql_info->name;
	int lost_connection=0;

	load_posfile(master);


	ulong len;
	Exit_status retval;
    mysql_init(&master->mysql);
	DEBUG("%s : mysql_init ok",sid);
	if ( connect_mysql(&master->mysql, master->mysql_info) != 0 ) {
		DEBUG("%s:fail to connect to master(tried 10 times), exit thread",sid);
	    mysql_close(&master->mysql);
		pthread_exit(NULL);
	}
	DEBUG("%s:ok to connect master", sid);
 
    NET *net;
    net = &(master->mysql.net);
   
    MYSQL *mm;
    mm = mysql_init(NULL);
    connect_mysql(mm, master->mysql_info);
    net =  &mm->net;


	master->mysql_info->server_id = get_server_id(mm);
	if ( master->mysql_info->server_id == -1) {
		DEBUG("%s : error get server_id",sid);
        EXIT_THREAD(mm);
	}

	/*check version*/
	if ((retval= check_master_version(master, mm)) != OK_CONTINUE) {
		DEBUG("%s:MySQL version error",sid);
        EXIT_THREAD(mm);
	}

    uint32_t server_id = generate_server_id(master->mysql_info->server_id);

	/*send COM_BINLOG_DUMP command*/
	if (send_binlog_dump(mm, master, server_id) != 0) {
		DEBUG("%s:Send binlog dump request error",sid);
        EXIT_THREAD(mm);
	}

	int i=0;

	DEBUG("%s: server_id=%d(%d) binlog_cur=%s", sid, master->mysql_info->server_id, server_id, master->binlog_cur);

    int reconnected = 0;
    int end_binlog = 0;
    const char * error_msg;
    const char *buf;

    char path[PATH_MAX];
    char sql_str[ITEM_LEN];
    size_t nwritten;
    int is_trans = 0;
    int xid_event = 0;
    char trans_sql_str[ITEM_LEN];
    trans_sql_str[0] = '\0';
    int last_slot = -1;
    unsigned long long intval = 0;

    while (setting.running) {

		//while((len = cli_safe_read(&master->mysql)) != 5);
		len = cli_safe_read(mm);

		if (len == packet_error) {
			DEBUG("%s:Got error reading packet from server %s:%s",
                    sid, master->mysql_info->host, 
                    mysql_error(mm)
                   );
            if (master->mysql.net.last_errno == 29) {
                DEBUG("abort as error 29");
                abort();
            }
            lost_connection = 1;
				
		} else {
//            DEBUG("%s cli_safe_read ok: len=%d", sid, len); 
        }

		if (len < 8 && net->read_pos[0] == 255) {
            DEBUG("%s: received end packet from server, apparent master shutdown:%s",
                    sid, mysql_error(&master->mysql));
            lost_connection = 1;
        }

		len--;
		buf = (const char*) net->read_pos + 1;

		if (lost_connection == 1) {
	        mysql_close(mm);
			mm = mysql_init(NULL);
            net = &mm->net;

			if ((reconnect_master(master, mm)) != 0) {
				DEBUG("%s:fail to reconnect to master",sid);
                EXIT_THREAD(mm);
			}
			else {
				DEBUG("%s:reconnect ok",sid);
				if ((retval= check_master_version(master, mm)) != OK_CONTINUE) {
					DEBUG("%s:MySQL version error",sid);
                    EXIT_THREAD(mm);
				}

				if (send_binlog_dump(mm, master, server_id) != 0) {
					DEBUG("%s:Send binlog dump request error",sid);
                    EXIT_THREAD(mm);
				}
				lost_connection = 0;
                reconnected = 1;
				continue;
			}
		}


        ulonglong pos_offset = uint4korr(buf + LOG_POS_OFFSET);
        if (master->trace_log_pos_offset > pos_offset && 
            reconnected && 
            pos_offset ||
            pos_offset == 0 && 
            master->trace_log_pos_offset != 54321) {
            if (setting.debug_sql) { 
                DEBUG("ignore the re-transfered events: %ld, %ld, reconn=%d",
                master->trace_log_pos_offset, pos_offset, reconnected);
            }
            continue;
        }
        end_binlog = 0;


		if (buf[EVENT_TYPE_OFFSET] == ROTATE_EVENT) {
			Log_event *ev;
			if (!(ev= Log_event::read_log_event(buf	,len , &error_msg , master->description_event))) {
				DEBUG("%s: could not construct log event object: %s", sid,error_msg);
                EXIT_THREAD(mm);
			}

			Rotate_log_event *rev= (Rotate_log_event *)ev;

			if (ev->when == 0) {
                if (strcmp(master->binlog_cur, rev->new_log_ident)) {
                    // update binlog fd and the content of poslog fd
                    free((void*)master->binlog_cur);
                    master->binlog_cur = strdup(rev->new_log_ident);
                    update_posfile_binlog(master);
                    
                }

                if (master->fd != -1)
                    close_binlog(master->fd);

                memset(path, 0, PATH_MAX);

                PRINTF_BINLOG_PATH(path, master);
                strcat(path,"/");
                strcat(path,master->binlog_cur);
                master->fd = open_binlog(path);
                DEBUG("%s: start GET NEW Binlog : %s", sid, master->binlog_cur);

                delete ev;
                continue;
			}
			else {
				delete ev;
                end_binlog = 1;
                pos_offset = 54321;
                reconnected = 0;

                DEBUG("%s: end of binlog: %s", sid, master->binlog_cur);
            }
        } else //if (setting.apply_binlog == 1) {
            if (buf[EVENT_TYPE_OFFSET] == QUERY_EVENT) {
                Query_log_event *query_ev = new Query_log_event(buf, 
                        len, 
                        master->description_event, QUERY_EVENT);
                assert((query_ev->db_len != 0));

                sql_str[0] = '\0';

                //support:
                //multi-database and multi-table
                //not support:
                //NULL database and NULL table
                //as for NULL situation, use could speed up replication by setting them

                if (master->mysql_info->database == NULL) {
                }

                else if (master->mysql_info->database != NULL && 
                        (my_strncmp(strdup(master->mysql_info->database), 
                                    query_ev->db, 
                                    query_ev->db_len) == 0)) {

                    if (query_ev->is_trans_keyword() &&
                        setting.autocommit == 0) {

                        snprintf(trans_sql_str, ITEM_LEN, "%s;", query_ev->query);
                        is_trans = 1;

                    }else {
                        assert(master->mysql_info->database != NULL);

                        // TODO: for join mode, two tables appeared in query
                        // then which slot should be found?
                        int slot = find_slot_by_table(master, query_ev->query);
                        if (slot != -1) {
                           
                            char *inc_convert_query = NULL;
                            if (setting.autoinc_id && intval != 0 && 
                                (strstr(query_ev->query, "INSERT ") || strstr(query_ev->query, "insert "))) {
                                inc_convert_query = convert_autoinc_id_with_val(query_ev->query, strdup(setting.autoinc_id), intval);
                                intval = 0;
                            }

                            int to_free = 0;
                            char *tp = transform_query(inc_convert_query==NULL?query_ev->query:inc_convert_query,
                                    query_ev->db, 
                                    mysql_info->database, 
                                    glob_master[slot].mysql_info->table,
                                    &to_free);

                            if (is_trans) {
                                put_data_to_buffer(slot, trans_sql_str);
                                is_trans = 0;
                                trans_sql_str[0] = '\0';
                            } 
                            last_slot = slot;

                            if (tp != NULL)
                                put_data_to_buffer(slot, tp);
                            else
                                put_data_to_buffer(slot, inc_convert_query==NULL?query_ev->query:inc_convert_query);

                            if (to_free) free(tp);
                            if (inc_convert_query) free(inc_convert_query);
                        }

                    } 
               } else {
                    //  master db is not NULL but not match query event's db
                    delete query_ev;
                    continue;
                }

                delete query_ev;
            }

            if (buf[EVENT_TYPE_OFFSET] == XID_EVENT &&
                (setting.autocommit == 0)) {
                if (1 && last_slot != -1)
                    put_data_to_buffer(last_slot, "commit");
#if 0
                Xid_log_event *xid_ev = new Xid_log_event(buf, master->description_event);
                delete xid_ev;
#endif
            }

   
            
        if (buf[EVENT_TYPE_OFFSET] == INTVAR_EVENT) {
            #if 1
            Intvar_log_event *ev = new Intvar_log_event(buf, master->description_event);
            intval = ev->val;
            delete ev;
            #endif
        } 


        if (buf[EVENT_TYPE_OFFSET] == FORMAT_DESCRIPTION_EVENT) {
            delete master->description_event;
            master->description_event= new Format_description_log_event(buf, len, master->description_event);
        }

        if (setting.save_binlog == 1) {
            nwritten = write(master->fd, buf, len);

            if (nwritten == -1) {
                DEBUG("write binlog file %d %s error:%s", master->fd, master->binlog_cur, strerror(errno));
                EXIT_THREAD(mm);
            }
            else if (nwritten != len) {
                DEBUG("write binlog short:%s", strerror(errno));
                EXIT_THREAD(mm);
            }
        } 

        // trace the LOG_POS_OFFSET to posfil
        //
        if (setting.debug_sql) {
            DEBUG("trace log_pos_offset shift: %d, %d", master->trace_log_pos_offset, uint4korr(buf + LOG_POS_OFFSET));
        }

        master->trace_log_pos_offset = pos_offset;
        
        /*
        if (master->trace_log_pos_events++ >= EVENT_INTERVAL &&
            (master->trace_log_pos_events % EVENT_INTERVAL == 0)) {
            master->trace_log_pos_offset =uint4korr(buf + LOG_POS_OFFSET);
            // TODO: do we really need this?
            //update_posfile_binlog(master);
        }
        */


        memset((void*)buf, 0, len); 
	}

	mysql_close(&master->mysql);
	pthread_exit(0);
}


/*------------------------------------------------------------------------------
 * 
 * about binlog
 *
 *----------------------------------------------------------------------------*/

void get_log_dir(const char*name, char * path, size_t len) {

	snprintf(path, len, "%s/%s", setting.base_dir, name);
}


/* open binlog file */
static int open_binlog(const char *binlog_path) {

	int fd;
	struct stat s;

	/*
	   open the binlog as APPEND,if not exist ,create it
	 */
	if ((fd = open(binlog_path, O_WRONLY|O_CREAT|O_TRUNC, 0644)) == -1) {
		DEBUG("open binlog file %s error:%s",  binlog_path, strerror(errno));
		return -1;
	}

	if (fstat(fd, &s) != 0) {
		DEBUG("fstat error:%s", strerror(errno));
		close(fd);
		return -1;
	}

	//the head of binlog is a MAGIC NUM
	if (s.st_size == 0) {
		/* write the binlog magic number */
		if (4 != write(fd, BINLOG_MAGIC, 4)) {
			DEBUG("write binlog magic error:%s", strerror(errno));
			return -1;
		}
	}

	return fd;
}

/* write the binlog to file */
static size_t write_binlog(master_info_st * master, const char * buf,
                            size_t len, my_off_t pos) {
	size_t nwritten;
	assert(master->fd != -1);
	nwritten = write(master->fd, buf, len);

	if (nwritten == -1) {
		DEBUG("write binlog file error:%s", strerror(errno));
		return -1;
	}
	else if (nwritten != len) {
		DEBUG("write binlog short:%s", strerror(errno));
		return nwritten;
	}

	return len;
}

/* close the binlog file */
static int close_binlog(int fd) {

	assert(fd != -1);

	if (fsync(fd)) {
		DEBUG("fsync binlog %d error:%s", fd, strerror(errno));
		return -1;
	}

	if (close(fd)) {
		DEBUG("binlog close error:%s", strerror(errno));
		return -1;
	}
	return 0;
}


int daemon() {
	umask(0);
	int fd, len;

	switch (fork()) {
		case -1:
			DEBUG("fork() failed:%s", strerror(errno));
			return -1;
		case 0: /* child */
			break;
		default: /* parent */
			exit(0);
	}
	setsid();

	if (chdir(setting.base_dir)<0) {
		DEBUG("chdir error\n");
		return -1;
	}
	printf("\n\n");
	DEBUG("###################starts###################");

	if ((fd = open(setting.pidfile, O_RDWR|O_CREAT|O_SYNC|O_TRUNC, 0644)) == -1) {
		DEBUG("open pidfile error:%s", strerror(errno));
		exit(0);
	}
	char buf[INT64_LEN + 1];
	len = snprintf(buf, INT64_LEN + 1, "%llu", (uint64_t)getpid());
	if (write(fd, buf, len) != len) {
		DEBUG("write error:%s", strerror(errno));
		return -1;
	}
	if (close(fd)) {
		DEBUG("close error:%s", strerror(errno));
		return -1;
	}

	return 0;
}


int init_setting() 
{

    setting.save_binlog  = 0;
    setting.daemon       = 0;
    setting.part_binlog  = 0;
    setting.forward_slave = 0;
    setting.apply_binlog = 0;
    setting.debug_sql    = 0;
    setting.autocommit   = 0;
    setting.pattern      = 1;
    setting.ignore       = NULL;
    setting.dump_diff    = 0;
    setting.force_group  = 0;
    setting.autoinc_id   = NULL;
    setting.running      = 1;

    return 0;
}

int init_variables() {
	pthread_mutex_init(&repsync_logfile_mutex,NULL);

    char cnf_com[PATH_MAX];
    snprintf(cnf_com, MAX_STRING_LEN, "%s/%s", setting.base_dir, "rep.cnf");
    setting.cnf_file = strdup(cnf_com);

    snprintf(cnf_com, MAX_STRING_LEN, "%s/%s", setting.base_dir, "/"PACKAGE".pid");
    setting.pidfile = strdup(cnf_com);

    snprintf(cnf_com, MAX_STRING_LEN, "%s/%s", setting.base_dir, "/"PACKAGE".log");
    setting.logfile = strdup(cnf_com);


    slave_conn_mutex = (pthread_mutex_t*) calloc(setting.n_slave, sizeof(pthread_mutex_t));

    int i = 0;
    for (i=0; i<setting.n_slave; i++) {
        glob_slave[i].mysql = (MYSQL*) calloc(glob_slave[i].n_master, sizeof(MYSQL)); //FIXME: free it
        int j = 0;
        for (j=0; j<glob_slave[i].n_master; j++) {
            mysql_init(&glob_slave[i].mysql[j]);
            connect_mysql(&glob_slave[i].mysql[j], glob_slave[i].mysql_info);
            glob_slave[i].mysql_info->server_id = get_server_id(&glob_slave[i].mysql[j]);
        }

        pthread_mutex_init(&slave_conn_mutex[i], NULL);
    }

    

   	if (mkdir(setting.base_dir, 0777) == -1 && errno!=EEXIST) {
		DEBUG("create repsync dir error : %s\n",setting.base_dir);
		return -1;
	}
    
    if ((fp_stdout=freopen(setting.logfile,"a",stdout))==NULL) {
		printf("error freopen\n");
		return -1;
	}

    sems_full= (sem_t*)malloc(sizeof(sem_t)* setting.n_master);
    sems_empty= (sem_t*)malloc(sizeof(sem_t)*setting.n_master);
 
    node = (node_t*) malloc(sizeof(node_t) * setting.n_master);
    buffer = (char***)malloc(sizeof(char**) * setting.n_master);
    for (i=0; i<setting.n_master; i++) {
        sem_init(&sems_full[i], 0, SLOT_ITEMS);
        //sem_init(&sems_full[i], 0, 0);
        sem_init(&sems_empty[i],0, 0);
        node[i].head=node[i].tail=node[i].loops=0;

        buffer[i] = (char**) malloc(sizeof(char*) * SLOT_ITEMS);
        int j = 0;
        for (j=0; j<SLOT_ITEMS; j++)
            buffer[i][j] = (char*) malloc(ITEM_LEN);
    }

    // init pk_map from autoinc_id
    if (setting.autoinc_id) {
        pk_map = (pk_map_t*) malloc(sizeof(pk_map_t) * MAX_PK);
        memset(pk_map, 0, MAX_PK*sizeof(pk_map_t));
        int pk_id = 0;
        char *ps = NULL;
        char *search_table, *search_pk;
        char *str2, *ptok, *ptoken, *psubtoken;
        char *pk_pair = strdup(setting.autoinc_id);
        while ((ptok = strtok_r(pk_pair, ",", &ptoken))) {
            pk_pair = NULL;
            str2 = ptok;
            search_table=strtok_r(str2, ":", &psubtoken);
            str2=NULL;
            search_pk=strtok_r(str2, ":", &psubtoken);

            int table_len = strlen(search_table);
            char *low_table = (char*)malloc(table_len+1);
            low_table[table_len] = '\0';
            char *upp_table = (char*)malloc(table_len+1);
            upp_table[table_len] = '\0';
            for (int i=0; i<table_len; i++) {
                low_table[i] = tolower(search_table[i]);
                upp_table[i] = toupper(search_table[i]);
            }

            int pk_len = strlen(search_pk);
            char *low_pk = (char*)malloc(pk_len+1);
            low_pk[pk_len] = '\0';
            char *upp_pk = (char*)malloc(pk_len+1);
            upp_pk[pk_len] = '\0';
            for (int j=0; j<pk_len; j++) {
                low_pk[j] = tolower(search_pk[j]);
                upp_pk[j] = toupper(search_pk[j]);
            }

            pk_map[pk_id].upp_table = upp_table;
            pk_map[pk_id].low_table = low_table;
            pk_map[pk_id].upp_pk    = upp_pk;
            pk_map[pk_id].low_pk    = low_pk;
            pk_map[pk_id].valid     = 1;
            pk_id++;
        }
    }

    return 0;
}

/* open the pos file for the master */
static int open_create_file(const char * path, bool create, const char *name) {
	int fd;
	char file[PATH_MAX]; 
	int flags;
	snprintf(file, PATH_MAX, "%s/%s", path, name);

	flags = create?O_RDWR|O_CREAT:O_RDWR;
	if ((fd = open(file, flags, 0777)) == -1) {
		return -1;
	}

	return fd;
}


/* update the posfile content */
static int update_posfile_binlog(master_info_st * master) {
	uint32_t n;
	assert(master->pos_fd != -1);

	if (lseek(master->pos_fd, SEEK_SET, 0) != 0) {
		DEBUG("lseek error:%s", strerror(errno));
		return -1;
	}

	size_t len=strlen(master->binlog_cur);
	n = htonl(len);
	if (write(master->pos_fd, (const void *)&n, POSFILE_STR_LEN_SIZE) != POSFILE_POS_SIZE) {
		DEBUG("write the posfile string len error:%s", strerror(errno));
		return -1;
	}
	/* write the string not include the null */
	if (write(master->pos_fd, (const void *)master->binlog_cur, len) != len) {
		DEBUG("write the posfile string error:%s", strerror(errno));
		return -1;
	}

	if (fsync(master->pos_fd) != 0) {
		DEBUG("fsync error:%s", strerror(errno));
		return -1;
	}	

	return 0;
}


static int load_posfile(master_info_st *master) {

	char path[PATH_MAX];

    PRINTF_BINLOG_PATH(path, master);
    
	if (mkdir(path, 0777) == -1) {
		if (errno != EEXIST) {
			DEBUG("mkdir error:%s", strerror(errno));
			return -1;
		}
	}


    master->fd       = -1;
    master->dump_fd  = NULL;

	int fd;
    
	if ( (fd=open_create_file(path, false, "posfile") ) == -1) {	//posfile doesnot exist
		if (master->binlog_start == NULL)
			master->binlog_cur = strdup("");
		else {
			master->binlog_cur = strdup(master->binlog_start);
            master->fd = open_create_file(path, true, master->binlog_cur);
        }


		fd = open_create_file(path, true, "posfile");	//create posfile
		master->pos_fd = fd;

		update_posfile_binlog(master);

    } else {
		uint32_t n, len;
		char name[PATH_MAX];
		master->pos_fd = fd;

		if (lseek(master->pos_fd, SEEK_SET, 0) != 0) {
			DEBUG("lseek error:%s", strerror(errno));
			return -1;
		}

		if (read(master->pos_fd, (void *)&n, POSFILE_STR_LEN_SIZE) != POSFILE_POS_SIZE) {
			DEBUG("read the posfile string len error:%s", strerror(errno));
			return -1;
		}
		len = ntohl(n);

		if (read(master->pos_fd, (void *)name, len) != len) {
			DEBUG("read the posfile string error:%s", strerror(errno));
			return -1;
		}
		name[len] = 0;

		master->binlog_cur = strdup(name);
        
        PRINTF_BINLOG_PATH(path, master);
        strcat(path, "/");
        strcat(path, master->binlog_cur);
        //master->fd = open_binlog(path);
    }
    
	DEBUG("%s:binlog_cur=%s",master->mysql_info->name,master->binlog_cur);
   
    // handle the table sql dumping
    PRINTF_DUMPSQL_PATH(path, master);
    
    master->dump_fd = fopen(path,"w"); // open(path, O_RDWR/*|O_APPEND*/|O_CREAT, 0777);
    if (master->dump_fd == NULL) {
        DEBUG("failed to create dump file:%s",  path);
    }

	return 0;
}

void usage() {
    printf("%s\n%s\n%s\n%s\n%s\n\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n",
    "DRC for mysql version: 0.1.0",
    "usage:",
    "    ./repsync -f rep.cnf -x -b -D",
    " or ./repsync -f rep.cnf -b -D",
    " or ./repsync -f rep.cnf -x -d",
    "arguments:",
    "   -x               apply binlog",                        
    "   -d               debug                                                      ",
    "   -D               daemon                                                     ",
    "   -f <cnf.file>    specify config file, -f rep.cnf                            ",
    "   -k               kill the app                                               ",
    "   -p               part recevied and apply binlog                             ",
    "   -b               save binlog on local                                       ",
    "   -e               exactly match the table name, default using pattern        ",
    "   -g               force group in duplicate master scenarios                  ",
    "   -i <tbl:pk>,<tbl:pk>  insert with primary key for autoincreament            ",
    "   -a               autocommit for transaction                                 ",
    "   -s               service provides (must dump diff and might forward querys) ",
    "   -t               forward query to slave to speedup slave replication(temp solution) ",
    "   -h               help");
}




int get_options(int argc, char *argv[]) {

    int c;
    while ((c = getopt (argc, argv, "axbdeghDkpstf:v:i:")) != -1)
        switch (c)
        {
            case 'x':
                setting.apply_binlog = 1;
                break;
            case 'a':
                setting.autocommit = 1;
                break;
            case 'd':
                setting.debug_sql = 1;
                break;
            case 'D':
                setting.daemon = 1;
                break;
            case 'f':
                setting.cnf_file = optarg;
                break;
            case 'g':
                setting.force_group = 1;
                break;
            case 'i':
                setting.autoinc_id = optarg;
                break;
            case 't':
                setting.forward_slave = 1;
                break;
            case 'k':
                repsync_kill();
                abort();
                break;
            case 'p':
                setting.part_binlog = 1;
                break;
            case 'e':
                setting.pattern = 0;
                break;
            case 'b':
                setting.save_binlog = 1;
                break;
            case 'v':
                setting.ignore = optarg;
                break;

            case 'c':
                setting.test = atoi(optarg);
                break;

            case 's':
                setting.dump_diff = 1;
                break;

            case 'h':
                usage();
                exit(0);
                break;
            case '?':
                if (optopt == 'f')
                    fprintf (stderr, "Option -%c requires an argument.\n", optopt);
                else if (isprint (optopt))
                    fprintf (stderr, "Unknown option `-%c'.\n", optopt);
                else
                    fprintf (stderr,
                            "Unknown option character `\\x%x'.\n",
                            optopt);
                usage();
                abort();
            default:
//                repsync_kill();
                abort ();
        }

        return 0;
}



int main(int argc, char * argv[]) {
	/*
	   ignore SIGHUP signal
	 */
	struct sigaction sa;
	sa.sa_handler=SIG_IGN;
	sigemptyset(&sa.sa_mask);
	sa.sa_flags=0;
	if (sigaction(SIGHUP,&sa,NULL)<0)
		printf("sig error:SIGHUP\n");
	if (sigaction(SIGHUP,&sa,NULL)<0)
		printf("sig error:SIGHUP\n");

    init_setting();
    get_options(argc, argv);


	/*
	   check if repsync is already running
	 */
	FILE *fp;
	char path[PATH_MAX];
    snprintf(path, PATH_MAX, "%s/%s", setting.base_dir, "/"PACKAGE".pid");
	if ((fp=fopen(path,"r")) != NULL) {
		char buf[60]="/proc/";
		char pid[60];
		if (fgets(pid,60,fp)==NULL) {
			DEBUG("error while fgets");
			return -1;
		}	

		strcat(buf,pid);

		DIR *dp;
		if ((dp=opendir(buf))!=NULL) {
			printf("repsync is already running\n");
            closedir(dp);
			exit(0);
		}
		else {
			printf("repsync is not running,run it\n");
        }
	}
    
	if (load_cnf_info() != 0)
		repsync_clean(0);
    
    if (init_variables() != 0) return -1;
 
    if (setting.daemon)
	    daemon();


    if (mysql_library_init(0, NULL, NULL)) {
        fprintf(stderr, "could not initialize MySQL library\n");
        exit(1);
    }

	int i, k;
    int loops = 0;
	for (i = 0; i != setting.n_master; ) {
        if ((glob_master+i)->group_index == 0)
            loops = 0;
        (glob_master+i)->tid_index = loops++;
		create_master_thread(glob_master+i);
        i += (glob_master+i)->n_instance;
	}
    for (i = 0; i != setting.n_master; i++) {
		create_slave_thread(glob_master+i);
	}
 
    for (i = 0; i != setting.n_slave; i++) {
        for (k=0; k<glob_slave[i].n_slides; k++) {
            pthread_join(glob_slave[i].tids[k], NULL);
        }
	}
	for (i = 0; i != setting.n_master; i++) {
		pthread_join(glob_master[i].tid, NULL);
	}

    //mysql_library_end();

	repsync_clean(0);

	return 0;
}

